/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hop.pipeline.transforms.databaselookup.readallcache;

import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.row.IValueMeta;

import java.util.BitSet;

class EqIndex extends Index {

  static Index nonEqualityIndex(int column, IValueMeta valueMeta, int rowsAmount) {
    return new EqIndex(column, valueMeta, rowsAmount, true);
  }

  private final boolean isMatchingNonEquality;

  EqIndex(int column, IValueMeta valueMeta, int rowsAmount) {
    this(column, valueMeta, rowsAmount, false);
  }

  private EqIndex(int column, IValueMeta valueMeta, int rowsAmount, boolean isMatchingNonEquality) {
    super(column, valueMeta, rowsAmount);
    this.isMatchingNonEquality = isMatchingNonEquality;
  }

  @Override
  void doApply(SearchingContext context, IValueMeta lookupMeta, Object lookupValue)
      throws HopException {
    int firstValue = findInsertionPointOf(new IndexedValue(lookupValue, -1));
    final int length = values.length;
    if (firstValue == length || valueMeta.compare(values[firstValue].key, lookupValue) != 0) {
      // nothing was found
      if (isMatchingNonEquality) {
        // everything is acceptable, just do nothing
        return;
      }
      context.setEmpty();
    } else {
      BitSet bitSet = context.getWorkingSet();
      bitSet.set(values[firstValue].row, true);
      int lastValue = firstValue + 1;
      while (lastValue != length && valueMeta.compare(values[lastValue].key, lookupValue) == 0) {
        bitSet.set(values[lastValue].row, true);
        lastValue++;
      }

      context.intersect(bitSet, isMatchingNonEquality);
    }
  }

  @Override
  int getRestrictionPower() {
    // "==" is a good restriction, whereas "!=" is quite weak
    return isMatchingNonEquality ? Short.MAX_VALUE : Short.MIN_VALUE;
  }
}
